package com.ruoyi.job.thread;

import com.alibaba.fastjson.JSONObject;
import com.api.equipment.RemoteEquipmentService;
import com.api.equipment.domain.LgEquipmentError;
import com.api.equipment.domain.LgMachine;
import com.api.equipment.domain.LgTemperature;
import com.ruoyi.common.core.utils.SpringUtils;
import com.ruoyi.job.redis.RedisUtil;
import com.ruoyi.job.util.MailUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: lek
 * @Description: TODO
 * @Date: 2025/3/16 16:41
 * @Version: 1.0
 */

public class MacThread extends Thread {
    private String serial;
    private String cmd;
    private Random random = new Random();
    String brokerUrl = "tcp://124.223.12.211:1883";
    String pubclientId = "machine_publisher";
    String subclientId = "machine_subscribe";
    String topic = "sensor/machine/";
    int pubQos = 0;
    int subQos = 1;
    public static ConcurrentHashMap<String, Thread> consoleThreadMap = new ConcurrentHashMap<String, Thread>();
    public ConcurrentHashMap<String, JSONObject> subMessage = new ConcurrentHashMap<String, JSONObject>();

    private volatile boolean running = true; // 新增运行状态标志
    private MqttClient publisherClient;
    private MqttClient subscriberClient;
    private MailUtil mailUtil = SpringUtils.getBean(MailUtil.class);

    private RemoteEquipmentService remoteEquipmentService = SpringUtils.getBean(RemoteEquipmentService.class);

    RedisUtil redisUtil = SpringUtils.getBean(RedisUtil.class);

    public String getSerial() {
        return serial;
    }

    public void setSerial(String serial) {
        this.serial = serial;
    }

    public static ConcurrentHashMap<String, Thread> getConsoleThreadMap() {
        return consoleThreadMap;
    }

    public static void setConsoleThreadMap(ConcurrentHashMap<String, Thread> consoleThreadMap) {
        MacThread.consoleThreadMap = consoleThreadMap;
    }

    public String getCmd() {
        return cmd;
    }

    public void setCmd(String cmd) {
        this.cmd = cmd;
    }

    public ConcurrentHashMap<String, JSONObject> getSubMessage() {
        return subMessage;
    }

    public void setSubMessage(ConcurrentHashMap<String, JSONObject> subMessage) {
        this.subMessage = subMessage;
    }

    @Override
    public void run() {
        try {
            if ("publish".equals(cmd)) {
                consoleThreadMap.put("publish" + serial, Thread.currentThread());
                publish();
            } else if ("subscribe".equals(cmd)) {
                consoleThreadMap.put("subscribe" + serial, Thread.currentThread());
                subscribe();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 根据命令类型移除对应的线程引用
            if ("publish".equals(cmd)) {
                consoleThreadMap.remove("publish" + serial);
            } else if ("subscribe".equals(cmd)) {
                consoleThreadMap.remove("subscribe" + serial);
            }
            if(redisUtil != null){
                //jedisPool.returnResourceObject(jedis);  // 已废弃，推荐使用jedis.close()方法
                redisUtil.close(redisUtil.getJedis());// 关闭redis连接,还回redis连接池中；
            }
        }
    }

    // 发布方法改进
    public void publish() throws MqttException {
        try {
            publisherClient = new MqttClient(brokerUrl, pubclientId + "_" + serial);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            publisherClient.connect(options);
            String targetTopic = topic + serial;
            while (running && !Thread.currentThread().isInterrupted()) {
                // 构造并发送消息
                JSONObject sensorData = new JSONObject();
                double temperature = random.nextInt(5) + random.nextDouble();
                sensorData.put("temperature", temperature);
                sensorData.put("humidity", 50 + random.nextInt(40));
                MqttMessage message = new MqttMessage(sensorData.toJSONString().getBytes());
                message.setQos(pubQos);
                publisherClient.publish(targetTopic, message);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    running = false;
                    break;
                }
            }
        } finally {
            // 仅关闭发布客户端
            if (publisherClient != null && publisherClient.isConnected()) {
                try {
                    publisherClient.disconnectForcibly(1000);
                    publisherClient.close();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // 订阅方法改进
    public void subscribe() throws MqttException {
        try {
            subscriberClient = new MqttClient(brokerUrl, subclientId + "_" + serial);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            subscriberClient.connect(options);
            subscriberClient.subscribe(topic + serial, subQos);


            subscriberClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
//                System.out.println("Connection lost: " + cause.getMessage());
                    running = false; // 触发终止
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    // 处理消息逻辑...
                    String payload = new String(message.getPayload());
                    JSONObject data = JSONObject.parseObject(payload); // Fastjson解析
                    LgMachine lgMachine = remoteEquipmentService.getInfo(serial);
                    lgMachine.setTemperature(data.getInteger("temperature"));
                    lgMachine.setHumidity(data.getInteger("humidity"));

                    remoteEquipmentService.update(lgMachine);

                    //判断当前温湿度是否合规  不合规存入异常表
                    if (data.getInteger("temperature") > lgMachine.getLgColdType().getMaxTemp()  || data.getInteger("temperature") < lgMachine.getLgColdType().getMinTemp()) {
                        LgEquipmentError lgEquipmentError = new LgEquipmentError();
                        lgEquipmentError.setDeviceNumber(lgMachine.getDeviceNumber());
                        lgEquipmentError.setAlarmTime(new Timestamp(new Date().getTime()).toString());
                        lgEquipmentError.setAlarmMessage("温度异常");
                        lgEquipmentError.setAlarmCount(1);
                        lgEquipmentError.setEquipmentType("2");
                        lgEquipmentError.setIsProcessed(0);
                        remoteEquipmentService.insertEquipError(lgEquipmentError);

                        System.out.println(lgMachine.getSysUser().getEmail());
                        //给管理员发送邮件
                        mailUtil.sendSimpleMail(lgMachine.getSysUser().getEmail(), "温度异常", "温度异常");
                    }
                    if (data.getInteger("humidity") > lgMachine.getLgColdType().getMaxHumidity() || data.getInteger("humidity") < lgMachine.getLgColdType().getMinHumidity()){
                        LgEquipmentError lgEquipmentError = new LgEquipmentError();
                        lgEquipmentError.setDeviceNumber(lgMachine.getDeviceNumber());
                        lgEquipmentError.setAlarmTime(new Timestamp(new Date().getTime()).toString());
                        lgEquipmentError.setAlarmMessage("湿度异常");
                        lgEquipmentError.setAlarmCount(1);
                        lgEquipmentError.setEquipmentType("2");
                        lgEquipmentError.setIsProcessed(0);
                        remoteEquipmentService.insertEquipError(lgEquipmentError);
                        System.out.println(lgMachine.getSysUser().getEmail());
                        //给管理员发送邮件
                        mailUtil.sendSimpleMail(lgMachine.getSysUser().getEmail(), "温度异常", "温度异常");
                    }
                    List temp = redisUtil.getList("equipment:machine:temperature" + serial);
                    List humidity = redisUtil.getList("equipment:machine:humidity" + serial);
                    List time = redisUtil.getList("equipment:machine:time" + serial);

                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
                    String timeString = sdf.format(new Date());
                    //将数据存入到redis
                    if (temp == null && humidity == null && time == null) {
//                    System.out.println("创建缓存");
                        List templist = new ArrayList();
                        templist.add(data.getInteger("temperature"));
                        redisUtil.setList("equipment:machine:temperature" + serial, templist);
                        List humiditylist = new ArrayList();
                        humiditylist.add(data.getInteger("humidity"));
                        redisUtil.setList("equipment:machine:humidity" + serial, humiditylist);
                        List timelist = new ArrayList();
                        timelist.add(timeString);
                        redisUtil.setList("equipment:machine:time" + serial, timelist);
                    } else {
//                    System.out.println("已有缓存 读取缓存");
                        if (temp.size() < 10) {
//                        System.out.println("缓存小于10 添加");
                            temp.add(data.getInteger("temperature"));
                            redisUtil.setList("equipment:machine:temperature" + serial, temp);
                            humidity.add(data.getInteger("humidity"));
                            redisUtil.setList("equipment:machine:humidity" + serial, humidity);
                            time.add(timeString);
                            redisUtil.setList("equipment:machine:time" + serial, time);
                        } else {
//                        System.out.println("缓存大于10 删除");
                            temp.remove(0);
                            temp.add(data.getInteger("temperature"));
                            redisUtil.setList("equipment:machine:temperature" + serial, temp);
                            humidity.remove(0);
                            humidity.add(data.getInteger("humidity"));
                            redisUtil.setList("equipment:machine:humidity" + serial, humidity);
                            time.remove(0);
                            time.add(timeString);
                            redisUtil.setList("equipment:machine:time" + serial, time);
                        }
                        System.out.println("缓存数据" + serial + "：" + temp + " " + humidity + " " + time);
                    }
                    //将获取到的数据 进行判断 判断温湿度是否合适  如果与设备设置的值不匹配 则发送邮件通知

//                System.out.println("\nReceived Sensor Data:");
//                System.out.println("Temperature: " + data.getDouble("temperature") + "℃");
//                System.out.println("Humidity: " + data.getInteger("humidity") + "%");
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                }
            });

            // 循环监听终止条件
            while (running && !Thread.currentThread().isInterrupted()) {
                try {
                    Thread.sleep(1000); // 允许中断的休眠
                } catch (InterruptedException e) {
                    // 重新设置中断标志并退出循环
                    Thread.currentThread().interrupt();
                    running = false;
                    break;
                }
            }
        } finally {
            // 确保关闭订阅客户端
            if (subscriberClient != null && subscriberClient.isConnected()) {
                try {
//                    subscriberClient.unsubscribe(topic + serial);
//                    subscriberClient.disconnectForcibly(1000);
                    subscriberClient.close();
                } catch (MqttException e) {
                    System.out.println("Error closing subscriber client: " + e.getMessage());
                }
            }
        }
    }
}
